#include "MessageQueueManager.h"



void MessageQueueManager::SAsyncRecvDataCallback(uS::Async* handle)
{
	if (handle)
	{
		MessageQueueManager* pX = (MessageQueueManager*)(handle->getData());
		if (pX)
		{
			pX->MemAsyncRecvDataCallback();
		}
	}
}

MessageQueueManager::MessageQueueManager()
{
}


MessageQueueManager::~MessageQueueManager()
{
}

void MessageQueueManager::InitMessageQueueManager(std::shared_ptr<ProcManager> proc)
{
	_procPt = proc;
	_dispathcThread = std::make_shared<std::thread>(&MessageQueueManager::DispathMessageThreadFunc, this);
}

void MessageQueueManager::AddWebRequestMessage(SessionType sessionid, const char* msg, size_t len)
{
	auto msg2 = std::shared_ptr<SessionRequestMessage>(new SessionRequestMessage{sessionid, {msg, len} });
	_webRecvQueue.enqueue(msg2);
	if (_run)
	{
		_asyncRecvClientData->send();
	}
}

void MessageQueueManager::AddWebResponseMessage(SessionType sessionid, const char* msg, size_t len, ResponseType broadType)
{
	auto msg2 = std::shared_ptr<SessionResponseMessage>(new SessionResponseMessage({ ResponseType::unicast ,sessionid, {msg, len} }));
	_webSendQueue.enqueue(msg2);
}

std::shared_ptr<SessionResponseMessage> MessageQueueManager::GetWebMessageResp()
{
	std::shared_ptr<SessionResponseMessage> msgRet = std::shared_ptr<SessionResponseMessage>(new SessionResponseMessage({ ResponseType::unicast, INVALID_SESSION_ID, {""} }));
	_webSendQueue.try_dequeue(msgRet);
	return msgRet;
}

void MessageQueueManager::AddSocketMessage(SessionType sessionid, char* msg, size_t len)
{

}

void MessageQueueManager::DispathMessageThreadFunc()
{
	uWS::Hub hub;
	_asyncRecvClientData = std::make_shared<uS::Async>(hub.getLoop());
	_asyncRecvClientData->setData(this);
	_asyncRecvClientData->start(&MessageQueueManager::SAsyncRecvDataCallback);
	_run = true;
	hub.run();
}

void MessageQueueManager::MemAsyncRecvDataCallback()
{
	
	size_t maxSize{ 10 };
	std::vector<std::shared_ptr<SessionRequestMessage>> vecMsgs;
	//vecWebMsgs.resize(maxSize);
	auto webmsgsize = _webRecvQueue.try_dequeue_bulk(std::back_inserter(vecMsgs), maxSize);
	while (webmsgsize > 0)
	{
		for (auto& msg : vecMsgs)
		{
			_procPt->DispatchClientMessage(msg);
		}
		vecMsgs.clear();
		webmsgsize = _webRecvQueue.try_dequeue_bulk(std::back_inserter(vecMsgs), maxSize);
	}
	vecMsgs.clear();
	auto sockmsgsize = _sockRecvQueue.try_dequeue_bulk(std::back_inserter(vecMsgs), maxSize);
	while (webmsgsize > 0)
	{
		for (auto& msg : vecMsgs)
		{
			_procPt->DispatchClientMessage(msg);
		}
		vecMsgs.clear();
		webmsgsize = _sockRecvQueue.try_dequeue_bulk(std::back_inserter(vecMsgs), maxSize);
	}

}
